package com.zjs.jsp.fdservice.messagelistener;


import com.currencyboot.service.rocketmq.messagelistener.Interface.MessageListenerConsumerInterface;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zjs.jsp.domain.amap.Address;
import com.zjs.jsp.domain.config.Param;
import com.zjs.jsp.domain.jsp.Jsp_ExceptionOrder;
import com.zjs.jsp.domain.jsp.Jsp_Order;
import com.zjs.jsp.domain.jsp.Jsp_Task;
import com.zjs.jsp.fdservice.implement.AdditionalDataJspTask;
import com.zjs.jsp.fdservice.service.ConfigService;
import com.zjs.jsp.fdservice.service.NewJspTask;
import com.zjs.jsp.fdservice.service.SendJspTask;
import com.zjs.jsp.fdservice.service.amap.GeoCodingService;
import com.zjs.jsp.fdservice.service.mongodb.ExceptionOrderExpand;
import com.zjs.jsp.fdservice.service.mongodb.JspTaskRepository;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.bson.types.ObjectId;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * <p>Title: ConsumableMessageListenerConsumer </p>
 * <p>@Description:  内销    </p>
 * <p>Company:  </p>
 *
 * @author 李文
 * @date 2016年9月25日 上午10:06:08
 */
public class ConsumableMessageListenerConsumer implements MessageListenerConsumerInterface
{

    private static final Logger LOGGER = LoggerFactory
            .getLogger(ConsumableMessageListenerConsumer.class);

    private static final ObjectMapper jsonMapper = new ObjectMapper();


    @Value("${ampkey}")
    private String ampkey;

    @Autowired
    private ExceptionOrderExpand exoExpand;


    @Autowired
    private JspTaskRepository jspTaskRepository;

    @Autowired
    private SendJspTask sendJspTask;

    @Autowired
    private NewJspTask newJspTask;

    @Autowired
    private ConfigService configService;

    @Autowired
    private RedissonClient client;

    @Autowired
    private GeoCodingService geoCodingService;

    /**
     * 操作状态
     */
    @Value("${czzt}")
    private Integer czzt;

    /**
     * 追加订单数据
     */
    @Autowired
    private AdditionalDataJspTask additionalData;


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(String strBody,
                                                    MessageExt msg,
                                                    ConsumeConcurrentlyContext context) {


        // 当前订单数据
        Jsp_Order jspOrder;
        try {
            // 进行数据整理
            jspOrder = jsonMapper.readValue(strBody, Jsp_Order.class);
        } catch (Exception e) {
            LOGGER.error("  分单服务  解析 客户订单 MQ 数据异常  数据是： " + strBody, e);
            Jsp_ExceptionOrder zxpExceptionOrder = new Jsp_ExceptionOrder("分单服务 解析 MQ 数据异常", strBody, 1);
            exoExpand.save(zxpExceptionOrder);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        try {
            return getConsumeConcurrentlyStatus(strBody, jspOrder);
        } catch (Exception e) {
            exoExpand.saveEx(jspOrder, e, jspOrder.getOuuid() + " 未知异常  数据是： " + strBody,
                    10, "未知异常");
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }

    }

    private ConsumeConcurrentlyStatus getConsumeConcurrentlyStatus(String strBody, Jsp_Order jspOrder) {

        //   获取配置信息   缓存查询
        Param param;
        String Pk;
        Date ZDSX;
        String paramDate = "";
        String corpFdInfo;
        try {
            param = configService.getParamList();
            if (StringUtils.isEmpty(jspOrder.getMdbm())) {
                throw new Exception(" 没有门店编码 ");
            }
            Pk = configService.getPkCorpByShopCode(jspOrder.getMdbm());
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, -param.getZDSX());
            ZDSX = calendar.getTime();
            paramDate = param.toString();
            if (StringUtils.isEmpty(Pk)) {
                throw new Exception(" 没有厅点编码 ");
            }
            corpFdInfo = configService.getCorpFdInfo(Pk);
        } catch (Exception e) {
            exoExpand.saveEx(jspOrder, e, jspOrder.getOuuid() + " 获取配置信息异常  数据是： " + strBody + "  获取的参数是  " + paramDate, 2, "获取配置信息");
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }


        // 必须条件  坐标匹配    与  人工分单 不进行和包
        ConsumeConcurrentlyStatus x = getConsumeConcurrentlyStatus(strBody, jspOrder, param, Pk, paramDate, corpFdInfo);
        if (x != null) {
            return x;
        }

        List<Jsp_Task> jspTasks = jspTaskRepository.findByClientFlagAndMdbmAndDateGreaterThanAndCzztNotAndFdlx(jspOrder.getClientFlag(), jspOrder.getMdbm(), ZDSX, czzt, 0);

        // 没有历史遗留任务包
        if (jspTasks == null || jspTasks.size() == 0) {
            // 创建新的任务包
            if (newJspTask.CreateNewJspTask(jspOrder, Pk, param, strBody)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } else {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }

        // 获取分类后的任务 Map
        LinkedHashMap<String, List<Jsp_Task>> mapJspTask = getMapJspTask(jspTasks, jspOrder);
        if (mapJspTask == null) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        //是否追加和包成功
        Boolean isAdditional = getaBoolean(strBody, jspOrder, param, Pk, mapJspTask);
        // 没有和包成功
        if (!isAdditional) {
            // 创建新的任务包
            LOGGER.info(" {} 和包失败创建新服务 ", jspOrder.getOuuid());
            if (!newJspTask.CreateNewJspTask(jspOrder, Pk, param, strBody)) {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }

        //  处理 未下发的数据
        sendMQ(strBody, jspOrder, mapJspTask.get("wxfJspTask"));
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    /**
     * 必须条件  坐标匹配    与  人工分单 不进行和包
     */
    private ConsumeConcurrentlyStatus getConsumeConcurrentlyStatus(String strBody, Jsp_Order jspOrder, Param param, String pk, String paramDate, String corpFdInfo) {
        try {
            if (StringUtils.isEmpty(jspOrder.getSra()) || StringUtils.isEmpty(jspOrder.getSrlx())) {
                Address address = geoCodingService.getGeocode(jspOrder.getSra(), ampkey);
                // 判断是否成功  0 表示失败
                if ("0".equals(address.getStatus()) || address.getGeocodes() == null || address.getGeocodes().size() == 0) {
                    throw new Exception("获取高德地图坐标异常");
                } else {
                    //        获取  经纬度
                    String location = address.getGeocodes().get(0).getLocation();
                    String[] split = location.split(",");
                    //经度
                    jspOrder.setSrlx(new BigDecimal(split[0]));
                    //纬度
                    jspOrder.setSrly(new BigDecimal(split[1]));
                }
            }
        } catch (Exception e) {
            exoExpand.saveEx(jspOrder, e, jspOrder.getOuuid() + " 获取高德地图坐标异常 " + strBody + "  获取的参数是  " + paramDate, 2, "获取高德地图坐标异常");
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }

        // 人工分单 不进行和包
        if ("1".equals(corpFdInfo)) {
            // 创建新的任务包
            LOGGER.info("{} 人工分单 ", jspOrder.getOuuid());
            if (!newJspTask.CreateNewJspTaskRG(jspOrder, pk, param, strBody)) {
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            } else {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        }
        return null;
    }

    /**
     * 是否追加和包成功
     */
    private Boolean getaBoolean(String strBody, Jsp_Order jspOrder, Param param, String pk, LinkedHashMap<String, List<Jsp_Task>> mapJspTask) {
        try {
            for (Map.Entry<String, List<Jsp_Task>> entry : mapJspTask.entrySet()) {
                //追加订单逻辑
                if (additionalData.additionalOrder(jspOrder, pk, param.getJDSX(), param.getDDZHBJ(), entry.getValue(), strBody)) {
                    return true;
                }
            }
        } catch (Exception e) {
            exoExpand.saveEx(jspOrder, e, "和包方法出错 数据是： " + strBody, 7, "和包方法出错 异常 " + e.getMessage());
        }
        return false;
    }


    /**
     * 获取分类后的任务
     */
    private LinkedHashMap<String, List<Jsp_Task>> getMapJspTask(List<Jsp_Task> jspTasks, Jsp_Order jspOrder) {
        // 未下发任务包（等待任务包池）
        List<Jsp_Task> wxfJspTask = new ArrayList<>();
        // 已经下发过的任务包
        List<Jsp_Task> yjxfJspTask = new ArrayList<>();
        for (Jsp_Task jspTask : jspTasks) {
            //并发照成的订单多次出现   任务包中已经有该订单 将不处理
            for (Jsp_Order j : jspTask.getOd()) {
                if (j.getPoc().equals(jspOrder.getPoc())) {
                    LOGGER.info(jspOrder.getPoc() + "  该订单已经处理过 ");
                    return null;
                }
            }
            //筛选 还未下发任务包的数据
            if (jspTask.getXfzt() == 1) {
                yjxfJspTask.add(jspTask);
            } else {
                wxfJspTask.add(jspTask);
            }
        }
        // 有历史遗留任务包   追单
        LinkedHashMap<String, List<Jsp_Task>> mapJspTask = new LinkedHashMap<>();

        // 将其他需要追单的数据添加到  追单数据源中
        mapJspTask.putAll(additionalData.getJspTaskMap(yjxfJspTask));

        // 将未进行下发任务包  在任务池中的订单 添加到追单数据源中
        mapJspTask.put("wxfJspTask", wxfJspTask);
        return mapJspTask;
    }

    /**
     * 分布式锁 发送MQ 信息
     *
     * @param strBody
     * @param jspOrder
     * @param wxfJspTask
     */
    private void sendMQ(String strBody, Jsp_Order jspOrder, List<Jsp_Task> wxfJspTask) {
        for (Jsp_Task task : wxfJspTask) {
            //  当前时间减去任务包创建时间  是否大于 设置的等待时间
            RLock pkLock = client.getLock("id" + task.getId());
            // 判断厅点是否上锁  上锁就放弃
            if (pkLock.isLocked()) {
                continue;
            }
            try {
                pkLock.lock(2, TimeUnit.MINUTES);
                task = jspTaskRepository.findOne(new ObjectId(task.getId()));
                if (task.getXfzt() == 1) {
                    continue;
                }
                long minute = System.currentTimeMillis() - task.getDate().getTime();
                minute = minute / 1000;
                if (task.getOd().size() >= task.getZd() || minute >= task.getDdjs()) {
                    //  发送数据到MQ
                    sendJspTask.sendMqData(strBody, jspOrder, null, task);
                }
            } catch (Exception e) {

            } finally {
                // 解锁
                pkLock.unlock();
            }
        }
    }

}
